package cn.com.guage.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * 
* @version:1.0.1
* @Description: flink的demo程序,统计单词个数
* @author: yangdechao
* @date: datedate 2021年11月12日 下午3:15:22
 */
public class BatchWordCount {

    public static void main(String[] args) throws Exception{
        //1.设置分布式执行环境
        ExecutionEnvironment env=ExecutionEnvironment.getExecutionEnvironment();
        
        //2.得到输入数据
        DataSet<String> text=env.fromElements(
                "My name is Xiongjie",
                "hello,World",
                "i Like java",
                "i Like java",
                "i Like java",
                "This world is Colorful,but i don't have work"
        );
        text.print();
        DataSet<Tuple2<String,Integer>> counts =
                //3.将数据切分成行的形式，格式为键值对<word,1>
                text.flatMap(new LineSplitter())
                //4.以key值“0”聚合操作,并且以value值“1”求和
                .groupBy(0)
                .sum(1);

        //3.数据集打印结果
        counts.print();
        
    }

    public static final class LineSplitter implements FlatMapFunction<String,Tuple2<String,Integer>> {


        /**
		 * 
		 */
		private static final long serialVersionUID = 172314625762779689L;

		public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            //1.格式化数据，并且切分行,以非字母分割
            String[] tokens=value.toLowerCase().split("\\W+");

            //2.发射键值对
            for (String token:tokens){
                if(token.length() > 0 ){
                    out.collect(new Tuple2<String, Integer>("xj："+token,1));
                }
            }

        }
    }


}


